import java.sql.PreparedStatement
import java.text.SimpleDateFormat
import java.util.{Date, Properties}

import com.alibaba.fastjson.JSON
import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.connector.jdbc.{JdbcConnectionOptions, JdbcExecutionOptions, JdbcSink, JdbcStatementBuilder}
import org.apache.flink.streaming.api.{CheckpointingMode, TimeCharacteristic}
import org.apache.flink.streaming.api.environment.CheckpointConfig
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer011


case class maidian
(
    vip_level :String,
    viewpoint_height :String,
    url_query :String,
    url_path :String,
    transportation_costs :String,
    total_price_of_commodity :String,
    title :String,
    timezone_offset :String,
    supplier_name :String,
    supplier_id :String,
    store_name :String,
    store_id :String,
    site_language :String,
    site_id :String,
    share_type2 :String,
    share_type :String,
    share_content :String,
    second_commodity :String,
    second_category :String,
    screen_width :String,
    screen_height :String,
    scene :String,
    room_name :String,
    room_id :String,
    room_category :String,
    register_method :String,
    referrer :String,
    receiver_province :String,
    receiver_name :String,
    receiver_city :String,
    receiver_area :String,
    receiver_address :String,
    rebate_amount :String,
    quantity_unit :String,
    present_price :String,
    player_name :String,
    player_id :String,
    platform_type :String,
    payment_method :String,
    page_title :String,
    os_version :String,
    os :String,
    original_price :String,
    order_should_pay :String,
    order_id :String,
    order_amount :String,
    order_actual_amount :String,
    oa_level :String,
    number_of_integral :String,
    network_type :String,
    monetary_unit :String,
    monetary :String,
    model :String,
    member_level :String,
    manufacturer :String,
    login_method :String,
    login_id :String,
    local_time :String,
    live_duration :String,
    lib_version :String,
    lib_method :String,
    lib :String,
    is_vip :String,
    is_use_discount :String,
    is_success :String,
    is_quick_login :String,
    is_own_goods :String,
    is_new :String,
    is_login :String,
    is_first_time :String,
    is_first_day :String,
    is_default_confirm :String,
    is_cash_back :String,
    inviter_id :String,
    invisitor_id :String,
    integral_discount_amount :String,
    if_use_integral :String,
    group_type :String,
    group_number :String,
    group_id :String,
    friend_num :String,
    first_commodity :String,
    first_category :String,
    express_name :String,
    event_duration :String,
    event :String,
    entrance :String,
    enter_source :String,
    element_type :String,
    element_id :String,
    distributor_level :String,
    distributor_fans :String,
    distinct_id :String,
    distibute_content :String,
    discount_type :String,
    discount_price :String,
    discount_name :String,
    discount_amount :String,
    delivery_method :String,
    cost :String,
    contact_type :String,
    commodity_tag :String,
    commodity_spu :String,
    commodity_sku :String,
    commodity_size :String,
    commodity_score :String,
    commodity_quantity :String,
    commodity_name :String,
    commodity_id :String,
    commodity_discount_amount :String,
    commodity_detail_souce :String,
    commodity_colour :String,
    category_id :String,
    cash_back_name :String,
    cash_back_amount :String,
    cancel_reason :String,
    app_name :String,
    app_id :String,
    anonymous_id :String,
    anchor_type :String,
    activity_name :String,
    activity_id :String,
    account :String,
    time :String,
    dt :String
)


//重写JdbcStatementBuilder
class CHSinkBuilder extends JdbcStatementBuilder[maidian]{
  override def accept(t: PreparedStatement, u: maidian): Unit = {
    t.setString(1, u.vip_level)
    t.setString(2, u.viewpoint_height)
    t.setString(3, u.url_query)
    t.setString(4, u.url_path)
    t.setString(5, u.transportation_costs)
    t.setString(6, u.total_price_of_commodity)
    t.setString(7, u.title)
    t.setString(8, u.timezone_offset)
    t.setString(9, u.supplier_name)
    t.setString(10, u.supplier_id)
    t.setString(11, u.store_name)
    t.setString(12, u.store_id)
    t.setString(13, u.site_language)
    t.setString(14, u.site_id)
    t.setString(15, u.share_type2)
    t.setString(16, u.share_type)
    t.setString(17, u.share_content)
    t.setString(18, u.second_commodity)
    t.setString(19, u.second_category)
    t.setString(20, u.screen_width)
    t.setString(21, u.screen_height)
    t.setString(22, u.scene)
    t.setString(23, u.room_name)
    t.setString(24, u.room_id)
    t.setString(25, u.room_category)
    t.setString(26, u.register_method)
    t.setString(27, u.referrer)
    t.setString(28, u.receiver_province)
    t.setString(29, u.receiver_name)
    t.setString(30,u.receiver_city)
    t.setString(31,u.receiver_area)
    t.setString(32,u.receiver_address)
    t.setString(33,u.rebate_amount)
    t.setString(34,u.quantity_unit)
    t.setString(35,u.present_price)
    t.setString(36,u.player_name)
    t.setString(37,u.player_id)
    t.setString(38,u.platform_type)
    t.setString(39,u.payment_method)
    t.setString(40,u.page_title)
    t.setString(41,u.os_version)
    t.setString(42,u.os)
    t.setString(43,u.original_price)
    t.setString(44,u.order_should_pay)
    t.setString(45,u.order_id)
    t.setString(46,u.order_amount)
    t.setString(47,u.order_actual_amount)
    t.setString(48,u.oa_level)
    t.setString(49,u.number_of_integral)
    t.setString(50,u.network_type)
    t.setString(51,u.monetary_unit)
    t.setString(52,u.monetary)
    t.setString(53,u.model)
    t.setString(54,u.member_level)
    t.setString(55,u.manufacturer)
    t.setString(56,u.login_method)
    t.setString(57,u.login_id)
    t.setString(58,u.local_time)
    t.setString(59,u.live_duration)
    t.setString(60,u.lib_version)
    t.setString(61,u.lib_method)
    t.setString(62,u.lib)
    t.setString(63,u.is_vip)
    t.setString(64,u.is_use_discount)
    t.setString(65,u.is_success)
    t.setString(66,u.is_quick_login)
    t.setString(67,u.is_own_goods)
    t.setString(68,u.is_new)
    t.setString(69,u.is_login)
    t.setString(70,u.is_first_time)
    t.setString(71,u.is_first_day)
    t.setString(72,u.is_default_confirm)
    t.setString(73,u.is_cash_back)
    t.setString(74,u.inviter_id)
    t.setString(75,u.invisitor_id)
    t.setString(76,u.integral_discount_amount)
    t.setString(77,u.if_use_integral)
    t.setString(78,u.group_type)
    t.setString(79,u.group_number)
    t.setString(80,u.group_id)
    t.setString(81,u.friend_num)
    t.setString(82,u.first_commodity)
    t.setString(83,u.first_category)
    t.setString(84,u.express_name)
    t.setString(85,u.event_duration)
    t.setString(86,u.event)
    t.setString(87,u.entrance)
    t.setString(88,u.enter_source)
    t.setString(89,u.element_type)
    t.setString(90,u.element_id)
    t.setString(91,u.distributor_level)
    t.setString(92,u.distributor_fans)
    t.setString(93,u.distinct_id)
    t.setString(94,u.distibute_content)
    t.setString(95,u.discount_type)
    t.setString(96,u.discount_price)
    t.setString(97,u.discount_name)
    t.setString(98,u.discount_amount)
    t.setString(99,u.delivery_method)
    t.setString(100,u.cost)
    t.setString(101,u.contact_type)
    t.setString(102,u.commodity_tag)
    t.setString(103,u.commodity_spu)
    t.setString(104,u.commodity_sku)
    t.setString(105,u.commodity_size)
    t.setString(106,u.commodity_score)
    t.setString(107,u.commodity_quantity)
    t.setString(108,u.commodity_name)
    t.setString(109,u.commodity_id)
    t.setString(110,u.commodity_discount_amount)
    t.setString(111,u.commodity_detail_souce)
    t.setString(112,u.commodity_colour)
    t.setString(113,u.category_id)
    t.setString(114,u.cash_back_name)
    t.setString(115,u.cash_back_amount)
    t.setString(116,u.cancel_reason)
    t.setString(117,u.app_name)
    t.setString(118,u.app_id)
    t.setString(119,u.anonymous_id)
    t.setString(120,u.anchor_type)
    t.setString(121,u.activity_name)
    t.setString(122,u.activity_id)
    t.setString(123,u.account)
    t.setString(124,u.time)
    t.setString(125,u.dt)
  }
}

object FlinkSinkKafka {

  def main(args: Array[String]): Unit = {
    //获取流处理
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    //隐式转换
    import org.apache.flink.api.scala._
    //checkpoint配置
    env.enableCheckpointing(100);
    env.getCheckpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
    env.getCheckpointConfig.setMinPauseBetweenCheckpoints(500);
    env.getCheckpointConfig.setCheckpointTimeout(60000);
    env.getCheckpointConfig.setMaxConcurrentCheckpoints(1);
    env.getCheckpointConfig.enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);;

    //设置并行度
    env.setParallelism(1)

    //设置EventTime
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    //连接kafka
    val properties = new Properties()
    properties.setProperty("bootstrap.servers", "183.3.221.132:9092")  //183.3.221.132:9092
    properties.setProperty("group.id", "test")
    properties.setProperty("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    properties.setProperty("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

    //获取kafka生产者消息
    val stream = env.addSource(new FlinkKafkaConsumer011[String]("maidian-data", new SimpleStringSchema(), properties))
    //stream.print()

    val skinSql =
      """
        |INSERT INTO events(
        |       vip_level, viewpoint_height,  url_query,   url_path, transportation_costs,   total_price_of_commodity,  title, timezone_offset, supplier_name,  supplier_id,
        |       store_name, store_id,  site_language,  site_id, share_type2,   share_type,  share_content, second_commodity,  second_category, screen_width,
        |       screen_height, scene,  room_name,  room_id, room_category,   register_method,   referrer, receiver_province,  receiver_name, receiver_city,
        |       receiver_area,  receiver_address,  rebate_amount,   quantity_unit,   present_price,   player_name,  player_id, platform_type,  payment_method,    page_title,
        |       os_version,  os, original_price, order_should_pay,   order_id, order_amount, order_actual_amount, oa_level, number_of_integral,   network_type,
        |       monetary_unit, monetary, model,  member_level,  manufacturer, login_method,    login_id,  local_time,  live_duration,   lib_version,
        |       lib_method, lib,  is_vip,  is_use_discount,  is_success,  is_quick_login,  is_own_goods,  is_new,   is_login, is_first_time,
        |       is_first_day,  is_default_confirm, is_cash_back, inviter_id, invisitor_id,   integral_discount_amount,    if_use_integral,  group_type,   group_number,  group_id,
        |       friend_num,    first_commodity,  first_category,    express_name, event_duration, event,   entrance,  enter_source,   element_type,    element_id,
        |       distributor_level,  distributor_fans,  distinct_id,  distibute_content, discount_type,    discount_price,  discount_name,  discount_amount,     delivery_method,   cost,
        |       contact_type, commodity_tag,      commodity_spu, commodity_sku,    commodity_size,    commodity_score,    commodity_quantity,   commodity_name,       commodity_id,      commodity_discount_amount,
        |       commodity_detail_souce,   commodity_colour,  category_id,  cash_back_name,   cash_back_amount,  cancel_reason,   app_name,      app_id,   anonymous_id,        anchor_type,
        |       activity_name,  activity_id,  account, time, dt
        |) VALUES(
        |?,?,?,?,?,?,?,?,?,?,
        |?,?,?,?,?,?,?,?,?,?,
        |?,?,?,?,?,?,?,?,?,?,
        |?,?,?,?,?,?,?,?,?,?,
        |?,?,?,?,?,?,?,?,?,?,
        |?,?,?,?,?,?,?,?,?,?,
        |?,?,?,?,?,?,?,?,?,?,
        |?,?,?,?,?,?,?,?,?,?,
        |?,?,?,?,?,?,?,?,?,?,
        |?,?,?,?,?,?,?,?,?,?,
        |?,?,?,?,?,?,?,?,?,?,
        |?,?,?,?,?,?,?,?,?,?,
        |?,?,?,?,?
        |)
        |""".stripMargin


    val DateStream = stream.map(line => {

        val str = line.replace("$", "")

        val time = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date)
        val dt = new SimpleDateFormat("yyyyMMdd").format(new Date)

        val distinct_id = JSON.parseObject(str).getString("distinct_id")
        val anonymous_id = JSON.parseObject(str).getString("anonymous_id")
        val event = JSON.parseObject(str).getString("event")
        val login_id = JSON.parseObject(str).getString("login_id")

        val libs = JSON.parseObject(str).getString("lib")
        val lib = JSON.parseObject(libs).getString("lib")
        val lib_method = JSON.parseObject(libs).getString("lib_method")
        val lib_version = JSON.parseObject(libs).getString("lib_version")

        val properties = JSON.parseObject(str).getString("properties")

        val vip_level = JSON.parseObject(properties).getString("vip_level")
        val viewpoint_height = JSON.parseObject(properties).getString("viewpoint_height")
        val url_query = JSON.parseObject(properties).getString("url_query")
        val url_path = JSON.parseObject(properties).getString("url_path")
        val transportation_costs = JSON.parseObject(properties).getString("transportation_costs")
        val total_price_of_commodity = JSON.parseObject(properties).getString("total_price_of_commodity")
        val title = JSON.parseObject(properties).getString("title")
        val timezone_offset = JSON.parseObject(properties).getString("timezone_offset")
        val supplier_name = JSON.parseObject(properties).getString("supplier_name")
        val supplier_id = JSON.parseObject(properties).getString("supplier_id")
        val store_name = JSON.parseObject(properties).getString("store_name")
        val store_id = JSON.parseObject(properties).getString("store_id")
        val site_language = JSON.parseObject(properties).getString("site_language")
        val site_id = JSON.parseObject(properties).getString("site_id")
        val share_type2 = JSON.parseObject(properties).getString("share_type2")
        val share_type = JSON.parseObject(properties).getString("share_type")
        val share_content = JSON.parseObject(properties).getString("share_content")
        val second_commodity = JSON.parseObject(properties).getString("second_commodity")
        val second_category = JSON.parseObject(properties).getString("second_category")
        val screen_width = JSON.parseObject(properties).getString("screen_width")
        val screen_height = JSON.parseObject(properties).getString("screen_height")
        val scene = JSON.parseObject(properties).getString("scene")
        val room_name = JSON.parseObject(properties).getString("room_name")
        val room_id = JSON.parseObject(properties).getString("room_id")
        val room_category = JSON.parseObject(properties).getString("room_category")
        val register_method = JSON.parseObject(properties).getString("register_method")
        val referrer = JSON.parseObject(properties).getString("referrer")
        val receiver_province = JSON.parseObject(properties).getString("receiver_province")
        val receiver_name = JSON.parseObject(properties).getString("receiver_name")
        val receiver_city = JSON.parseObject(properties).getString("receiver_city")
        val receiver_area = JSON.parseObject(properties).getString("receiver_area")
        val receiver_address = JSON.parseObject(properties).getString("receiver_address")
        val rebate_amount = JSON.parseObject(properties).getString("rebate_amount")
        val quantity_unit = JSON.parseObject(properties).getString("quantity_unit")
        val present_price = JSON.parseObject(properties).getString("present_price")
        val player_name = JSON.parseObject(properties).getString("player_name")
        val player_id = JSON.parseObject(properties).getString("player_id")
        val platform_type = JSON.parseObject(properties).getString("platform_type")
        val payment_method = JSON.parseObject(properties).getString("payment_method")
        val page_title = JSON.parseObject(properties).getString("page_title")
        val os_version = JSON.parseObject(properties).getString("os_version")
        val os = JSON.parseObject(properties).getString("os")
        val original_price = JSON.parseObject(properties).getString("original_price")
        val order_should_pay = JSON.parseObject(properties).getString("order_should_pay")
        val order_id = JSON.parseObject(properties).getString("order_id")
        val order_amount = JSON.parseObject(properties).getString("order_amount")
        val order_actual_amount = JSON.parseObject(properties).getString("order_actual_amount")
        val oa_level = JSON.parseObject(properties).getString("oa_level")
        val number_of_integral = JSON.parseObject(properties).getString("number_of_integral")
        val network_type = JSON.parseObject(properties).getString("network_type")
        val monetary_unit = JSON.parseObject(properties).getString("monetary_unit")
        val monetary = JSON.parseObject(properties).getString("monetary")
        val model = JSON.parseObject(properties).getString("model")
        val member_level = JSON.parseObject(properties).getString("member_level")
        val manufacturer = JSON.parseObject(properties).getString("manufacturer")
        val login_method = JSON.parseObject(properties).getString("login_method")

        val local_time = JSON.parseObject(properties).getString("local_time")
        val live_duration = JSON.parseObject(properties).getString("live_duration")
        val is_vip = JSON.parseObject(properties).getString("is_vip")
        val is_use_discount = JSON.parseObject(properties).getString("is_use_discount")
        val is_success = JSON.parseObject(properties).getString("is_success")
        val is_quick_login = JSON.parseObject(properties).getString("is_quick_login")
        val is_own_goods = JSON.parseObject(properties).getString("is_own_goods")
        val is_new = JSON.parseObject(properties).getString("is_new")
        val is_login = JSON.parseObject(properties).getString("is_login")
        val is_first_time = JSON.parseObject(properties).getString("is_first_time")
        val is_first_day = JSON.parseObject(properties).getString("is_first_day")
        val is_default_confirm = JSON.parseObject(properties).getString("is_default_confirm")
        val is_cash_back = JSON.parseObject(properties).getString("is_cash_back")
        val inviter_id = JSON.parseObject(properties).getString("inviter_id")
        val invisitor_id = JSON.parseObject(properties).getString("invisitor_id")
        val integral_discount_amount = JSON.parseObject(properties).getString("integral_discount_amount")
        val if_use_integral = JSON.parseObject(properties).getString("if_use_integral")
        val group_type = JSON.parseObject(properties).getString("group_type")
        val group_number = JSON.parseObject(properties).getString("group_number")
        val group_id = JSON.parseObject(properties).getString("group_id")
        val friend_num = JSON.parseObject(properties).getString("friend_num")
        val first_commodity = JSON.parseObject(properties).getString("first_commodity")
        val first_category = JSON.parseObject(properties).getString("first_category")
        val express_name = JSON.parseObject(properties).getString("express_name")
        val event_duration = JSON.parseObject(properties).getString("event_duration")
        val entrance = JSON.parseObject(properties).getString("entrance")
        val enter_source = JSON.parseObject(properties).getString("enter_source")
        val element_type = JSON.parseObject(properties).getString("element_type")
        val element_id = JSON.parseObject(properties).getString("element_id")
        val distributor_level = JSON.parseObject(properties).getString("distributor_level")
        val distributor_fans = JSON.parseObject(properties).getString("distributor_fans")
        val distibute_content = JSON.parseObject(properties).getString("distibute_content")
        val discount_type = JSON.parseObject(properties).getString("discount_type")
        val discount_price = JSON.parseObject(properties).getString("discount_price")
        val discount_name = JSON.parseObject(properties).getString("discount_name")
        val discount_amount = JSON.parseObject(properties).getString("discount_amount")
        val delivery_method = JSON.parseObject(properties).getString("delivery_method")
        val cost = JSON.parseObject(properties).getString("cost")
        val contact_type = JSON.parseObject(properties).getString("contact_type")
        val commodity_tag = JSON.parseObject(properties).getString("commodity_tag")
        val commodity_spu = JSON.parseObject(properties).getString("commodity_spu")
        val commodity_sku = JSON.parseObject(properties).getString("commodity_sku")
        val commodity_size = JSON.parseObject(properties).getString("commodity_size")
        val commodity_score = JSON.parseObject(properties).getString("commodity_score")
        val commodity_quantity = JSON.parseObject(properties).getString("commodity_quantity")
        val commodity_name = JSON.parseObject(properties).getString("commodity_name")
        val commodity_id = JSON.parseObject(properties).getString("commodity_id")
        val commodity_discount_amount = JSON.parseObject(properties).getString("commodity_discount_amount")
        val commodity_detail_souce = JSON.parseObject(properties).getString("commodity_detail_souce")
        val commodity_colour = JSON.parseObject(properties).getString("commodity_colour")
        val category_id = JSON.parseObject(properties).getString("category_id")
        val cash_back_name = JSON.parseObject(properties).getString("cash_back_name")
        val cash_back_amount = JSON.parseObject(properties).getString("cash_back_amount")
        val cancel_reason = JSON.parseObject(properties).getString("cancel_reason")
        val app_name = JSON.parseObject(properties).getString("app_name")
        val app_id = JSON.parseObject(properties).getString("app_id")
        val anchor_type = JSON.parseObject(properties).getString("anchor_type")
        val activity_name = JSON.parseObject(properties).getString("activity_name")
        val activity_id = JSON.parseObject(properties).getString("activity_id")
        val account = JSON.parseObject(properties).getString("account")


      maidian(vip_level, viewpoint_height,  url_query,   url_path, transportation_costs,   total_price_of_commodity,  title, timezone_offset, supplier_name,  supplier_id,
              store_name, store_id,  site_language,  site_id, share_type2,   share_type,  share_content, second_commodity,  second_category, screen_width,
              screen_height, scene,  room_name,  room_id, room_category,   register_method,   referrer, receiver_province,  receiver_name, receiver_city,
              receiver_area,  receiver_address,  rebate_amount,   quantity_unit,   present_price,   player_name,  player_id, platform_type,  payment_method,    page_title,
              os_version,  os, original_price, order_should_pay,   order_id, order_amount, order_actual_amount, oa_level, number_of_integral,   network_type,
              monetary_unit, monetary, model,  member_level,  manufacturer, login_method,    login_id,  local_time,  live_duration,   lib_version,
              lib_method, lib,  is_vip,  is_use_discount,  is_success,  is_quick_login,  is_own_goods,  is_new,   is_login, is_first_time,
              is_first_day,  is_default_confirm, is_cash_back, inviter_id, invisitor_id,   integral_discount_amount,    if_use_integral,  group_type,   group_number,  group_id,
              friend_num,    first_commodity,  first_category,    express_name, event_duration, event,   entrance,  enter_source,   element_type,    element_id,
              distributor_level,  distributor_fans,  distinct_id,  distibute_content, discount_type,    discount_price,  discount_name,  discount_amount,     delivery_method,   cost,
              contact_type, commodity_tag,      commodity_spu, commodity_sku,    commodity_size,    commodity_score,    commodity_quantity,   commodity_name,       commodity_id,      commodity_discount_amount,
              commodity_detail_souce,   commodity_colour,  category_id,  cash_back_name,   cash_back_amount,  cancel_reason,   app_name,      app_id,   anonymous_id,        anchor_type,
              activity_name,  activity_id,  account, time , dt
      )

    })
    DateStream.print()


    DateStream.addSink(
      JdbcSink.sink(
        skinSql,
        new CHSinkBuilder,
        new JdbcExecutionOptions.Builder()
          .withBatchSize(1)
          .build(),
        new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
          .withDriverName("ru.yandex.clickhouse.ClickHouseDriver")
          .withUrl("jdbc:clickhouse://106.14.31.66:8123/xiaoxiangai")
          .withUsername("default")
          .withPassword("Ssyt@118")
          .build()
      )
    )

    env.execute()
  }
}
